package com.rec.smile.rpc.registry.zookeeper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rec.smile.rpc.api.util.Constant;
import com.rec.smile.rpc.registry.RootRegistryApi;

public class ZooKeeperServiceDiscovery extends RootRegistryApi {

	private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceDiscovery.class);
	
	private CountDownLatch countDownLatch = new CountDownLatch(1);
	
	private volatile List<String> dataList = new ArrayList<String>();
	
	private String registryAddress;

	public ZooKeeperServiceDiscovery(String registryAddress) {
		super();
		this.registryAddress = registryAddress;
		
		ZooKeeper zooKeeper = connectServer();
		if(zooKeeper != null){
			watchNode(zooKeeper);
		}
	}

	private void watchNode(final ZooKeeper zooKeeper) {
		try {
			List<String> children = zooKeeper.getChildren(Constant.ZK_REGISTRY_PATH, new Watcher() {
				
				@Override
				public void process(WatchedEvent event) {
					if(event.getType() == Event.EventType.NodeChildrenChanged){
						watchNode(zooKeeper);
					}
				}
			});
			
			for (String node : children) {
				this.dataList.add(new String(zooKeeper.getData(Constant.ZK_REGISTRY_PATH+"/"+node, false, null)));
			}
			
			LOGGER.debug("node data {}", dataList);
		} catch (KeeperException | InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	private ZooKeeper connectServer() {
		ZooKeeper zooKeeper = null;
		try {
			zooKeeper = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
				
				@Override
				public void process(WatchedEvent event) {
					if(event.getState() == Event.KeeperState.SyncConnected){
						countDownLatch.countDown();
					}
				}
			});
			countDownLatch.await();
		} catch (IOException | InterruptedException e) {
			LOGGER.error("", e);
		}
		return zooKeeper;
	}

	@Override
	public String discover() {
		if(dataList.isEmpty())
			return null;
		return dataList.get(ThreadLocalRandom.current().nextInt(dataList.size()));
	}
	
	
}
